package com.viknix.threadpool.manager.server.dependence.monitor.publisher;

import cn.hutool.core.collection.ConcurrentHashSet;
import com.viknix.threadpool.manager.server.dependence.monitor.event.Event;
import com.viknix.threadpool.manager.server.dependence.monitor.event.SlowEvent;
import com.viknix.threadpool.manager.server.dependence.monitor.listeners.Listener;
import lombok.extern.slf4j.Slf4j;

import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

/**
 * @Author: Dongqi
 * @Date: 2021/11/15 19:35
 * @Version 1.0
 * @Description: 慢事件的默认共享事件发布者实现。
 */
@Slf4j
public class DefaultSharePublisher extends DefaultPublisher {

    /**
     * key：慢事件的类型；value：监听该慢事件的监听器集合
     */
    private final Map<Class<? extends SlowEvent>, Set<Listener>> subMappings = new ConcurrentHashMap<Class<? extends SlowEvent>, Set<Listener>>();

    private final Lock lock = new ReentrantLock();

    /**
     * 为慢事件注册监听器
     *
     * @param listener      慢事件监听器
     * @param subscribeType 订阅的慢事件类型
     */
    public void addSubscriber(Listener listener, Class<? extends Event> subscribeType) {

        Class<? extends SlowEvent> subSlowEventType = (Class<? extends SlowEvent>) subscribeType;
        // 用于添加到父类属性同步。
        listeners.add(listener);

        lock.lock();
        try {
            // 从 subMappings 集合中获取所有该慢事件类型的监听器集合
            Set<Listener> sets = subMappings.get(subSlowEventType);
            if (sets == null) {
                // 如果该慢事件还没有注册监听器，那么会构建一个集合用于存放监听该慢事件的监听器
                Set<Listener> newSet = new ConcurrentHashSet<Listener>();
                newSet.add(listener);
                // 将该集合注册到 subMappings 集合中
                subMappings.put(subSlowEventType, newSet);
                return;
            }
            sets.add(listener);
        } finally {
            lock.unlock();
        }
    }

    /**
     * 从 sharePublisher 中移除该事件的监听器
     *
     * @param listener      {@link Listener}
     * @param subscribeType subscribe event type, such as slow event or general event.
     */
    public void removeSubscriber(Listener listener, Class<? extends Event> subscribeType) {

        Class<? extends SlowEvent> subSlowEventType = (Class<? extends SlowEvent>) subscribeType;
        // 用于移除到父类的属性同步。
        listeners.remove(listener);

        lock.lock();
        try {
            Set<Listener> sets = subMappings.get(subSlowEventType);

            if (sets != null) {
                // 从 subMappings 集合中移除
                sets.remove(listener);
            }
        } finally {
            lock.unlock();
        }
    }

    /**
     * 依次回调监听器方法
     *
     * @param event 事件源对象
     */
    @Override
    public void receiveEvent(Event event) {

        final long currentEventSequence = event.sequence();
        // get subscriber set based on the slow EventType.
        final Class<? extends SlowEvent> slowEventType = (Class<? extends SlowEvent>) event.getClass();

        // Get for Map, the algorithm is O(1).
        Set<Listener> listeners = subMappings.get(slowEventType);
        if (null == listeners) {
            log.debug(">>> [😐 NotifyCenter] No subscribers for slow event {}", slowEventType.getName());
            return;
        }

        // Notification single event subscriber
        for (Listener listener : listeners) {
            // Whether to ignore expiration events
            if (listener.ignoreExpireEvent() && lastEventSequence > currentEventSequence) {
                log.debug(">>> [😐 NotifyCenter] the {} is unacceptable to this subscriber, because had expire",
                        event.getClass());
                continue;
            }

            // Notify single subscriber for slow event.
            notifySubscriber(listener, event);
        }
    }
}
